package com.atguigu.flink.sqlfunction;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Created by Smexy on 2023/2/6
 */
public class Demo9_WindowJoin
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());

        SingleOutputStreamOperator<WaterSensor> ds1 = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);

        SingleOutputStreamOperator<WaterSensor> ds2 = env
            .socketTextStream("hadoop103", 8889)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);

        /*
                开启基于事件时间的窗口

                ds1 join ds2 on ds1.xx = ds2.xx
         */
        ds1.join(ds2)
           //指定左边流中的数据用什么字段去关联
           .where(WaterSensor::getId)
           //= 第二个流中的字段
           .equalTo(WaterSensor::getId)
           .window(TumblingEventTimeWindows.of(Time.seconds(5)))
           //和process的运行原理是一样的，全量聚合。一个窗口调用一次
           .apply(new JoinFunction<WaterSensor, WaterSensor, String>()
           {
               @Override
               public String join(WaterSensor first, WaterSensor second) throws Exception {
                   return first + "====" + second;
               }
           })
           .print();



        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
